package com.flink.process.function;

import com.flink.datasource.UserSource;
import com.flink.entity.User;
import com.flink.window.WindowFunctionDemo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * 描述:
 * 系统处理时间 定时器函数触发机制
 *
 * @author yanzhengwu
 * @create 2022-08-07 16:59
 */
public class ProcessTimeTimerFunctionDemo {

    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);


        //TODO 无序流的watermark生成策略，这使用系统处理时间所以并未设置水位线
        DataStream<User> stream = env.addSource(new UserSource());
        //TODO 注册定时器必须在keyBy之后使用KeyedProcessFunction才可以运行，否则就得报错
        stream.keyBy(User::getName)
                .process(new KeyedProcessFunction<String, User, String>() {
                    @Override
                    public void processElement(User value, Context ctx, Collector<String> out) throws Exception {
                        long l = ctx.timerService().currentProcessingTime();

                        out.collect(ctx.getCurrentKey()+"->数据到达时间：" + new Timestamp(l));
                        //TODO 注册一个定时器 (根据分组的key 10秒统计一次,将当前处理时间 +上10秒，注意：不是事件时间而是系统处理时间)
                        ctx.timerService().registerProcessingTimeTimer(l + 10 * 1000L);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect(ctx.getCurrentKey()+"->定时器触发时间："+new Timestamp(timestamp));
                    }


                });





        env.execute();
    }
}
